package com.kitten.comment.consumer;

import com.github.phantomthief.collection.BufferTrigger;
import com.kitten.comment.constants.RedisKeyConstants;
import com.kitten.comment.domain.dataobject.CommentDO;
import com.kitten.comment.domain.mapper.CommentDOMapper;
import com.kitten.comment.model.bo.CommentHeatBO;
import com.kitten.comment.util.HeatCalculator;
import com.kitten.framework.common.utils.JsonUtils;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.assertj.core.util.Lists;
import org.assertj.core.util.Sets;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

import static com.kitten.comment.constants.MQConstants.TOPIC_COMMENT_HEAT_UPDATE;

/**
 * @author kitten
 */
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "kitten_group_" + TOPIC_COMMENT_HEAT_UPDATE,
        topic = TOPIC_COMMENT_HEAT_UPDATE
)
public class CommentHeatUpdateConsumer implements RocketMQListener<String> {
    @Resource
    private CommentDOMapper commentDOMapper;

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    private BufferTrigger<String> bufferTrigger = BufferTrigger.<String>batchBlocking()
            .bufferSize(5000)
            .batchSize(30)
            .linger(Duration.ofSeconds(2))  // 2s后聚合一次
            .setConsumerEx(this::consumeMessage)
            .build();

    @Override
    public void onMessage(String s) {
        bufferTrigger.enqueue(s);
    }

    private void consumeMessage(List<String> messages) {
        log.info("==> 【评论热度值计算】聚合消息, size: {}", messages.size());
        log.info("==> 【评论热度值计算】聚合消息, {}", JsonUtils.toJsonString(messages));

        Set<Long> commentIds = Sets.newHashSet();
        messages.forEach(message -> {
            try {
                Set<Long> list = JsonUtils.parseSet(message, Long.class);
                commentIds.addAll(list);
            } catch (Exception e) {
                log.error("", e);
            }
        });
        // 查询一级评论信息 (select id, level, parent_id, user_id, child_comment_total, like_total from comment where id in (commentIds))
        List<CommentDO> commentDOS = commentDOMapper.selectByCommentIds(commentIds.stream().toList());
        List<Long> ids = Lists.newArrayList();
        List<CommentHeatBO> commentHeatBOS = Lists.newArrayList();
        // 构建CommentHeatBO
        commentDOS.forEach(commentDO -> {
            Long commentId = commentDO.getId();
            // 被点赞数
            Long likeTotal = commentDO.getLikeTotal();
            // 被回复数
            Long childCommentTotal = commentDO.getChildCommentTotal();

            // 计算热度值
            BigDecimal heatNum = HeatCalculator.calculateHeat(likeTotal, childCommentTotal);
            ids.add(commentId);
            commentHeatBOS.add(CommentHeatBO.builder()
                    .id(commentId)
                    .heat(heatNum.doubleValue())
                    .noteId(commentDO.getNoteId())
                    .build());
        });
        // 批量更新评论热度值 (update t_comment set heat =
        // case id when id in (ids)
        // then heat_value end where id in (ids))
        int count = commentDOMapper.batchUpdateHeatByCommentIds(ids, commentHeatBOS);

        if (count == 0) {
            return;
        }
        // 更新Redis缓存热度值记录
        updateRedisHotComment(commentHeatBOS);
    }

    /**
     * 更新Redis中热点评论ZSET
     * @param commentHeatBOS
     */
    private void updateRedisHotComment(List<CommentHeatBO> commentHeatBOS) {
        Map<Long, List<CommentHeatBO>> noteIdAndBOListMap = commentHeatBOS.stream()
                .filter(commentHeatBO -> commentHeatBO.getHeat() > 0)
                .collect(Collectors.groupingBy(CommentHeatBO::getNoteId));

        // 循环每个笔记的热点评论
        noteIdAndBOListMap.forEach((noteId, commentHeatBOList) -> {
            // 构建热点评论 Redis Key
            String key = RedisKeyConstants.buildCommentListKey(noteId);

            DefaultRedisScript<Long> script = new DefaultRedisScript<>();
            // Lua 脚本路径
            script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/update_comment_heat.lua")));
            // 返回值类型
            script.setResultType(Long.class);

            // 构建执行 Lua 脚本所需的 ARGS 参数
            List<Object> args = Lists.newArrayList();
            commentHeatBOList.forEach(commentHeatBO -> {
                args.add(commentHeatBO.getId()); // Member: 评论ID
                args.add(commentHeatBO.getHeat()); // Score: 热度值
            });

            // 执行 Lua 脚本
            redisTemplate.execute(script, Collections.singletonList(key), args.toArray());
        });
    }

}
